package data.platform.merge.internal.command;

import data.platform.common.domain.MetricValue;
import data.platform.common.merge.*;
import data.platform.common.service.query.MetricTagQueryService;
import data.platform.merge.config.MetricMergeConfig;
import data.platform.merge.internal.cache.AggregateItemCache;
import data.platform.merge.internal.cache.MergeItemCache;
import data.platform.merge.model.MergeItem;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;

@Service
@Slf4j
@RequiredArgsConstructor
public class MergeService {

    @Value("${checkpoint.realTime}")
    private boolean realTime;

    final MetricMergeConfig metricMergeConfig;

    final MergeItemCache mergeItemCache;

    final AggregateItemCache aggregateItemCache;

    final MetricTagQueryService metricTagQueryService;

    public Mono<Integer> process(MetricValue metricValue) {
        // 流数据， 获取归并配置项
        String moType = metricValue.getTag().get("moc");
        List<MergeItem> mergeItemList = mergeItemCache.findMergeItem(metricValue.getMetric(), moType);
        // 获取指标的tagId
        Optional<Integer> tagOptional = metricTagQueryService.findTagId(metricValue.getTagJson());
        Mono<Integer> tagMono;
        if (tagOptional.isPresent()) {
            tagMono = Mono.just(tagOptional.get());
        } else {
            // 如果缓存不存在，延后再次获取
            tagMono = Mono.defer(() -> Mono.delay(Duration.ofMillis(3000))
                    .map(i -> metricTagQueryService.findTagId(metricValue.getTagJson()).get()));
        }
        return tagMono.zipWhen(tagId -> Mono.just(metricValue)
                .doOnNext(mv -> {
                    mergeItemList.forEach(mergeItem -> {
                        // 统计时间窗口
                        MergeTimeWindow mergeTimeWindow = MergeTimeWindow.getMergeTimeWindow(mergeItem.getMergePeriod(), metricValue.getEventTime());
                        // 根据算法选择聚集类
                        ArithmeticType arithmeticType = metricMergeConfig.getArithmetics().get(mergeItem.getArithmetic());
                        Map<Integer, AggregateItem> tagIdCache = aggregateItemCache.getTagIdCache(mergeTimeWindow, arithmeticType, mv.getMetric());
                        AggregateItem aggregateItem = null;
                        if (arithmeticType == ArithmeticType.COMMON) {
                            aggregateItem = tagIdCache.computeIfAbsent(tagId, k -> new CommonAggregateItem(mergeItem.getMergePeriod(), mergeTimeWindow));
                        } else if (arithmeticType == ArithmeticType.LOADPERCENT) {
                            aggregateItem = tagIdCache.computeIfAbsent(tagId, k -> new LoadPercentAggregateItem(mergeItem.getMergePeriod(), mergeTimeWindow));
                        } else if (arithmeticType == ArithmeticType.IFCROWDTIME) {
                            aggregateItem = tagIdCache.computeIfAbsent(tagId, k -> new IfCrowdTimeAggregateItem(mergeItem.getMergePeriod(), mergeTimeWindow, metricMergeConfig.getIfCrowdTimePercentThreshold()));
                        } else if (arithmeticType == ArithmeticType.FAULT) {
                            aggregateItem = tagIdCache.computeIfAbsent(tagId, k -> new FaultAggregateItem(mergeItem.getMergePeriod(), mergeTimeWindow));
                        }
                        if (Objects.nonNull(aggregateItem)) {
                            aggregateItem.compute(mv.getValue(), mv.getEventTime());
                        }
                        if (realTime) {
                            // 实时存储数据变更
                            aggregateItemCache.persistent(mergeTimeWindow, arithmeticType, mv.getMetric(), tagId, aggregateItem);
                        }
                    });
                }))
                .map(tuple2 -> 1);
    }

}
